package com.seventeen.mq.proxy.server.proxy.thread;

import com.seventeen.mq.proxy.common.constant.DelFlagConstants;
import com.seventeen.mq.proxy.common.utils.StringUtils;
import com.seventeen.mq.proxy.server.proxy.ext.BaseExtMqProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 生产者代理资源池
 *
 * @author seventeen
 */
public class ProducerPool extends Thread {
    private static final Logger LOGGER = LoggerFactory.getLogger(ProducerPool.class);
    private final Map<Long, BaseExtMqProducer> producerMap = new ConcurrentHashMap<>(16);
    private final LinkedBlockingQueue<BaseExtMqProducer> producerQueue;
    private volatile boolean toStop = false;

    public ProducerPool() {
        this.producerQueue = new LinkedBlockingQueue<>();
    }

    public BaseExtMqProducer get(Long producerProxyId) {
        return producerMap.get(producerProxyId);
    }

    public void pushQueue(BaseExtMqProducer producer) {
        producerQueue.add(producer);
    }

    public void toStop() {
        LOGGER.info("producerPool toStop!");
        this.toStop = true;
    }

    @Override
    public void run() {
        while (!toStop) {
            try {
                BaseExtMqProducer mqProducer = producerQueue.poll(3L, TimeUnit.SECONDS);
                if (mqProducer == null) {
                    continue;
                }
                BaseExtMqProducer oldMqProducer = producerMap.get(mqProducer.getId());
                if (oldMqProducer != null) {
                    if (DelFlagConstants.DELETED.equals(mqProducer.getDelFlag())) {
                        oldMqProducer.destroy();
                        producerMap.remove(mqProducer.getId());
                        LOGGER.info("proxy producerMap update {}", producerMap);
                        continue;
                    }

                    if (StringUtils.equalsIgnoreCase(mqProducer.getVersion(), oldMqProducer.getVersion())) {
                        continue;
                    } else {
                        oldMqProducer.destroy();
                        producerMap.remove(mqProducer.getId());
                        LOGGER.info("proxy producerMap update {}", producerMap);
                    }
                }

                if (DelFlagConstants.DELETED.equals(mqProducer.getDelFlag())) {
                    continue;
                }
                mqProducer.init();
                producerMap.put(mqProducer.getId(), mqProducer);
                LOGGER.info("proxy producerMap update {}", producerMap);
            } catch (Throwable e) {
                LOGGER.error("proxy producerMap update error {}, producerMap {}", e.getMessage(), producerMap, e);
            }

        }

        LOGGER.info("toStop is {}", toStop);
        if (toStop) {
            LOGGER.info("producerQueue clear!");
            producerQueue.clear();
            for (BaseExtMqProducer mqProducer : producerMap.values()) {
                mqProducer.destroy();
            }
        }
    }
}
